package com.dhcc;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.CountDownLatch;

import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.util.StopWatch;

@SpringBootTest(classes = RocketmqProducerApplication.class)
public class RocketmqProducerDemo04 {

	@Autowired
	private RocketMQTemplate rocketMQTemplate;
	
	/**
	 * 异步消息生产者
	 */
	@Test
	public void testProducerDemoOneWay() {
		DefaultMQProducer producer = null;
		try {
			// 1. 创建消息生产者, 指定生产者所属的组名
			producer = new DefaultMQProducer("sendOneWay-rocketmq-producer-group");
			// 2. 指定Nameserver地址
			producer.setNamesrvAddr("192.168.91.200:9876");
			// 3. 启动生产者
			producer.start();
			// 5. 发送消息
			StopWatch stopWatch = new StopWatch();
			stopWatch.start();
			for (int i = 0; i < 100; i++) {
				// 4. 创建消息对象，指定主题、标签和消息体
				Message msg = new Message("oneWayTopic",("单向消息发送" + i + "...").getBytes(StandardCharsets.UTF_8));
				producer.sendOneway(msg);
			}

			stopWatch.stop();
			long totalTimeMillis = stopWatch.getTotalTimeMillis();
			System.out.println("单向总耗时：" + totalTimeMillis);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	/**
	 * 异步消费消息
	 */
	@Test
	public void testConsumerDemoOneWay() {
		DefaultMQPushConsumer consumer = null;
		try {
			// 1、创建消费者，指定消费者组
			consumer = new DefaultMQPushConsumer("oneWayAsync-consumer-consumer-group");
			// 2、消费者设置nameserver地址
			consumer.setNamesrvAddr("192.168.91.200:9876");
			// 3、设置订阅主题，订阅某一个topic的消息
			consumer.subscribe("oneWayTopic", "*");
			CountDownLatch countDownLatch = new CountDownLatch(100);
			// 4、注册消息监听器，等待消息推送后接受消息
			consumer.registerMessageListener(new MessageListenerConcurrently() {
				@Override
				public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
					String threadName = Thread.currentThread().getName();
					for (int i = 0; i < msgs.size(); i++) {
						System.out.println(threadName + "--- 单向消息：" + new String(msgs.get(i).getBody()));
					}
					countDownLatch.countDown();
					return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
				}
			});
			// 5、启动消费者与MQ建立连接
			consumer.start();
			countDownLatch.await();
		} catch (Exception e) {
			// TODO: handle exception
		} finally {
			if(consumer != null) {
				// 6、关闭消费者
				consumer.shutdown();
			}
		}
	}

	@Test
	public void testProducerDemo02() {
		// 参数一: topic， 如果想添加tag 可以使用"topic:tag"的写法
		// 参数二: 消息内容
		SendResult sendResult = rocketMQTemplate.syncSend("dhcc-sync-tags", "这是一条同步消息");
		System.out.println(sendResult);
	}

}
